package ml

import base.BaseModel
import bean.HBaseMeta
import org.apache.spark.ml.{Pipeline, PipelineModel}
import org.apache.spark.ml.classification.{DecisionTreeClassificationModel, DecisionTreeClassifier}
import org.apache.spark.ml.evaluation.MulticlassClassificationEvaluator
import org.apache.spark.ml.feature.{StringIndexer, StringIndexerModel, VectorAssembler}
import org.apache.spark.sql.types.DoubleType
import org.apache.spark.sql.{Column, DataFrame, Dataset, Row, SparkSession, functions}

object USG extends BaseModel {
  override def setAppName: String = "USG"

  override def setFourId: String = "158"

  override def getNewTag(spark: SparkSession, fiveTagDF: DataFrame, HBaseDF: DataFrame): DataFrame = {

    import spark.implicits._
    import org.apache.spark.sql.functions._
    //本需求中需要用到两张表tbl_goods、tbl_orders，现在只有orders
    var tbl_orders = HBaseDF

    //读取tbl_goods
    val tbl_goods: DataFrame = spark.read.format("tools.HBaseDataSource")
      .option(HBaseMeta.ZKHOSTS, "192.168.10.20")
      .option(HBaseMeta.ZKPORT, "2181")
      .option(HBaseMeta.HBASETABLE, "tbl_goods")
      .option(HBaseMeta.FAMILY, "detail")
      .option(HBaseMeta.SELECTFIELDS, "cOrderSn,productType,ogColor")
      .load()

    //tbl_goods.show()

    //将订单表和商品表进行关联，得到用户消费商品所属类别和颜色
    val orders_goods: DataFrame = tbl_orders.join(tbl_goods,
      tbl_orders.col("orderSn") === tbl_goods.col("cOrderSn"))
      .select("memberId", "productType", "ogColor")

    //orders_goods.show()
    //和业务部门确定商品类型和颜色的性别特征
    val label: Column = functions
      .when('ogColor.equalTo("樱花粉")
        .or('ogColor.equalTo("白色"))
        .or('ogColor.equalTo("香槟色"))
        .or('ogColor.equalTo("香槟金"))
        .or('productType.equalTo("料理机"))
        .or('productType.equalTo("挂烫机"))
        .or('productType.equalTo("吸尘器/除螨仪")), 1) //女
      .otherwise(0) //男
      .alias("gender")

    //决策树算法需要的特征数据不能是字符串类型，我们需要将字符串转化为数字类型
    //这里的编号，最好是数据库中读取
    val color: Column = functions
      .when('ogColor.equalTo("银色"), 1)
      .when('ogColor.equalTo("香槟金色"), 2)
      .when('ogColor.equalTo("黑色"), 3)
      .when('ogColor.equalTo("白色"), 4)
      .when('ogColor.equalTo("梦境极光【卡其金】"), 5)
      .when('ogColor.equalTo("梦境极光【布朗灰】"), 6)
      .when('ogColor.equalTo("粉色"), 7)
      .when('ogColor.equalTo("金属灰"), 8)
      .when('ogColor.equalTo("金色"), 9)
      .when('ogColor.equalTo("乐享金"), 10)
      .when('ogColor.equalTo("布鲁钢"), 11)
      .when('ogColor.equalTo("月光银"), 12)
      .when('ogColor.equalTo("时尚光谱【浅金棕】"), 13)
      .when('ogColor.equalTo("香槟色"), 14)
      .when('ogColor.equalTo("香槟金"), 15)
      .when('ogColor.equalTo("灰色"), 16)
      .when('ogColor.equalTo("樱花粉"), 17)
      .when('ogColor.equalTo("蓝色"), 18)
      .when('ogColor.equalTo("金属银"), 19)
      .when('ogColor.equalTo("玫瑰金"), 20)
      .otherwise(0)
      .alias("color")


    //类型ID应该来源于字典表,这里简化处理
    val productType: Column = functions
      .when('productType.equalTo("4K电视"), 9)
      .when('productType.equalTo("Haier/海尔冰箱"), 10)
      .when('productType.equalTo("Haier/海尔冰箱"), 11)
      .when('productType.equalTo("LED电视"), 12)
      .when('productType.equalTo("Leader/统帅冰箱"), 13)
      .when('productType.equalTo("冰吧"), 14)
      .when('productType.equalTo("冷柜"), 15)
      .when('productType.equalTo("净水机"), 16)
      .when('productType.equalTo("前置过滤器"), 17)
      .when('productType.equalTo("取暖电器"), 18)
      .when('productType.equalTo("吸尘器/除螨仪"), 19)
      .when('productType.equalTo("嵌入式厨电"), 20)
      .when('productType.equalTo("微波炉"), 21)
      .when('productType.equalTo("挂烫机"), 22)
      .when('productType.equalTo("料理机"), 23)
      .when('productType.equalTo("智能电视"), 24)
      .when('productType.equalTo("波轮洗衣机"), 25)
      .when('productType.equalTo("滤芯"), 26)
      .when('productType.equalTo("烟灶套系"), 27)
      .when('productType.equalTo("烤箱"), 28)
      .when('productType.equalTo("燃气灶"), 29)
      .when('productType.equalTo("燃气热水器"), 30)
      .when('productType.equalTo("电水壶/热水瓶"), 31)
      .when('productType.equalTo("电热水器"), 32)
      .when('productType.equalTo("电磁炉"), 33)
      .when('productType.equalTo("电风扇"), 34)
      .when('productType.equalTo("电饭煲"), 35)
      .when('productType.equalTo("破壁机"), 36)
      .when('productType.equalTo("空气净化器"), 37)
      .otherwise(0)
      .alias("productType")

    //用户数据中的字符串转化为数字并添加标签
    val orders_goods_Int: DataFrame = orders_goods.select('memberId, productType, color, label)

    //3 标签处理
    val labelInt: StringIndexerModel = new StringIndexer()
      .setInputCol("gender")
      .setOutputCol("label")
      .fit(orders_goods_Int)

    //4 特征向量化
    val features: VectorAssembler = new VectorAssembler()
      .setInputCols(Array("productType", "color"))
      .setOutputCol("features")

    //5 实例化决策树
    val decisionTree: DecisionTreeClassifier = new DecisionTreeClassifier()
      .setFeaturesCol("features")
      .setPredictionCol("prediction")
      .setMaxDepth(5)

    //6 创建pipline
    val pipeline: Pipeline = new Pipeline().setStages(Array(labelInt, features, decisionTree))

    //7 将数据分为训练数据和测试数据
    val Array(trainData, testData): Array[Dataset[Row]] = orders_goods_Int.randomSplit(Array(0.8, 0.2))

    //8 使用pipline对训练数据进行训练，对测试数据进行测试
    val model: PipelineModel = pipeline.fit(trainData)
    val testDF: DataFrame = model.transform(testData)

    //9 查看模型准确度
    //    val evaluator: MulticlassClassificationEvaluator = new MulticlassClassificationEvaluator()
    //      .setLabelCol("label") //设置原始数据的label
    //      .setPredictionCol("prediction")//设置根据数据预测的结果
    //    val score: Double = evaluator.evaluate(testDF)
    // println(score)

    //10 决策树过程
    // val decisionTreeClassificationModel: DecisionTreeClassificationModel = model.stages(2).asInstanceOf[DecisionTreeClassificationModel]
    //println(decisionTreeClassificationModel.toDebugString)

    //11 对用户id进行分组 计算商品男性的百分比和女性的百分比
    //计算时需要使用所有数据
    //获取训练的80%数据
    val trainDF: DataFrame = model.transform(trainData)
    val manWomanALL: DataFrame = trainDF.union(testDF)
      .select('memberId,
        when('prediction === 0, 1).otherwise(0) as "man",
        when('prediction === 1, 1).otherwise(0) as "woman"
      ).groupBy("memberId")
      .agg(
        sum('man) cast (DoubleType) as "manSum",
        sum('woman) cast (DoubleType) as "womanSum",
        count('man) cast (DoubleType) as "all"
      )

    //五级标签转化为Map
    val fiveTagMap: Map[String, Long] = fiveTagDF.collect().map(row => {
      (row(0).toString, row(1).toString.toLong)
  }).toMap

    var getSex = udf((manSum: Double, womanSum: Double, all: Double) => {
      //计算男性商品百分比
      var manPercent = manSum / all
      //计算女性商品百分比
      var womanPercent = womanSum / all
      if (manPercent >= 0.6) {
        fiveTagMap.get("0")
      } else if (womanPercent >= 0.6) {
        fiveTagMap.get("1")
      } else {
        fiveTagMap.get("-1")
      }
    })
    val newTags: DataFrame = manWomanALL.select('memberId as "userId", getSex('manSum, 'womanSum, 'all) as "tagsId")
    newTags
  }

  //主程序
  def main(args: Array[String]): Unit = {
    exec()
  }
}
